package com.htzz.collection.config;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSONObject;
import com.htzz.collection.service.CollectionService;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

import javax.annotation.PostConstruct;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * mqtt 消费者类
 * @author xudezhi
 *
 */
@Slf4j
@Configuration
@ConditionalOnProperty(prefix = "mqtt",name = "use", havingValue = "true", matchIfMissing = false)
public class MqttSendConfig{

	ExecutorService pool = Executors.newFixedThreadPool(1000);

	MqttClient client;

	@Value("${mqtt.broker.serverUri}")
	private String MqttUrl;

	@Value("${mqtt.broker.username}")
	private String userName;
	@Value("${mqtt.broker.password}")
	private String password;


	private static SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");


	/**
	 * MQTT连接器选项
	 */
	@Bean
	public MqttConnectOptions getSenderMqttConnectOptions() {
		MqttConnectOptions options = new MqttConnectOptions();
		// 设置连接的用户名
		if (!userName.trim().equals("")) {
			options.setUserName(userName);
		}
		// 设置连接的密码
		options.setPassword(password.toCharArray());
		// 设置连接的地址
		options.setServerURIs(new String[]{MqttUrl});
		// 设置超时时间 单位为秒
		options.setConnectionTimeout(10);
		// 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送心跳判断客户端是否在线
		// 但这个方法并没有重连的机制
		options.setKeepAliveInterval(20);
		// 设置“遗嘱”消息的话题，若客户端与服务器之间的连接意外中断，服务器将发布客户端的“遗嘱”消息。
		return options;
	}


	/**
	 * MQTT客户端
	 */
	@Bean
	public MqttPahoClientFactory senderMqttClientFactory() {
		DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
		factory.setConnectionOptions(getSenderMqttConnectOptions());
		return factory;
	}

	/**
	 * MQTT信息通道（生产者）
	 */
	@Bean(name = "mqttOutboundChanne")
	public MessageChannel mqttOutboundChannel() {
		return new DirectChannel();
	}

	/**
	 * MQTT消息处理器（生产者）
	 */
	@Bean
	@ServiceActivator(inputChannel = "mqttOutboundChanne")
	public MessageHandler mqttOutbound() {
		MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(
				UUID.randomUUID().toString(),
				senderMqttClientFactory());
		messageHandler.setAsync(true);
		messageHandler.setDefaultTopic("/down/#");
		return messageHandler;
	}

}
